package com.zt.bigdata.spark.spark.broadcast

import java.util.Properties

import com.zt.bigdata.practise.kafka.basic.KafkaProperties
import com.zt.bigdata.template.spark.redis.RedisTemplate
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SparkSession

object BroadcastVar {
  val redis_addr = "127.0.0.1:6379"

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[2]")
      .appName("BroadcastVar")

      .getOrCreate()
    RedisTemplate.init(redis_addr)
    RedisTemplate.get().set("a", "zt")

    val data = List(1, 2, 3, 4, 5, 6)
    val bdata: Broadcast[List[Int]] = spark.sparkContext.broadcast(data)


    val rdd = spark.sparkContext.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9), 5)

    //    rdd.foreachPartition {
    //      r =>
    //        RedisTemplate.init("10.200.150.8:6868")
    //        val jedis: Jedis = RedisTemplate.get()
    //        println(jedis.get("a") + " " + jedis.hashCode())
    //    }

    val kafkaProducer: Broadcast[KafkaSink[String, String]] = {
      val kafkaProducerConfig = {
        val p = new Properties()
        p.setProperty("bootstrap.servers", KafkaProperties.KAFKA_SERVER_URL)
        p.setProperty("key.serializer", classOf[StringSerializer].getName)
        p.setProperty("value.serializer", classOf[StringSerializer].getName)
        p
      }
      spark.sparkContext.broadcast(KafkaSink[String, String](kafkaProducerConfig))
    }

    val redisClient: Broadcast[RedisClient] =
      spark.sparkContext.broadcast(RedisClient(redis_addr.split(":")(0), 6379))

    rdd.foreach(str => {
      kafkaProducer.value.send(KafkaProperties.TOPIC, str.toString)
      val jedis = redisClient.value.getResource()
      try {
        println(jedis.get("a") + " " + jedis.hashCode())
      } finally {
        jedis.close
      }
    })

    rdd.foreachPartition(r => {
      val jedis = redisClient.value.getResource()
      println(jedis.get("a") + " " + jedis.hashCode())
    })

  }
}
